package org.shj.spark.operator;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;

//
public class CoalesceOperator {

	public static void main(String[] args) {

		SparkConf conf = new SparkConf().setMaster("local").setAppName("CoalesceOperator");
		JavaSparkContext sc = new JavaSparkContext(conf);
		
		List<String> list = Arrays.asList("zhuyin01","zhuyin02","zhuyin03","zhuyin04","zhuyin05","zhuyin06",
				"zhuyin07","zhuyin08","zhuyin09","zhuyin10","zhuyin11","zhuyin12");
		JavaRDD<String> staffRDD = sc.parallelize(list, 6);
		
		//coalesce算子，功能是将RDD的partition的数量缩减（合并分区）或增加
		//将一定的数据压缩到更少的partition分区中去
		//使用场景： 很多时候在filter算子应用之后会使用coalesce算子优化一下
		//因为filter算子应用到RDD上面后，RDD对应的每个partition中的数据有的多有的少，产生
		//数据倾斜，换句话说就是可能有的partition里面就剩下一条数据，而有的partition中数据偏多
		//建议使用coalesce算子，从而让各个partition中的数据更加紧凑
		
		
		JavaRDD<String> staffRDD2 = staffRDD.mapPartitionsWithIndex(new Function2<Integer, Iterator<String>, Iterator<String>>(){
			private static final long serialVersionUID = -7067045146566393502L;

			@Override
			public Iterator<String> call(Integer index, Iterator<String> it) throws Exception {
				List<String> list = new ArrayList<String>();
				while(it.hasNext()){
					String staff = it.next();
					list.add("Dept[" + (index+1) +"]" + staff);
				}
				return list.iterator();
			}
			
		}, true);
		
		//RDD.collect()，它是在Driver端执行， 慎用，它会把数据加载到本地
		//RDD.foreach()，它是在集群中执行。
		for(String info : staffRDD2.collect()){
			System.out.println(info);
		}
		
		JavaRDD<String> staffRDD3 = staffRDD2.coalesce(3);
		
		JavaRDD<String> staffRDD4 = staffRDD3.mapPartitionsWithIndex(new Function2<Integer, Iterator<String>, Iterator<String>>(){
			private static final long serialVersionUID = -2777962664851020869L;

			public Iterator<String> call(Integer index, Iterator<String> it) throws Exception {
				List<String> list = new ArrayList<String>();
				while(it.hasNext()){
					String staff = it.next();
					list.add("Dept[" + (index+1) +"]" + staff);
				}
				return list.iterator();
			}
			
		}, true);
		
		for(String info : staffRDD4.collect()){
			System.out.println(info);
		}
		
		sc.close();
	}

}
